﻿using Confluent.Kafka;
using Common.Json;
using Common.MQ;
using Newtonsoft.Json.Linq;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Transactions;

namespace Common.Kafka
{
    /// <summary>
    /// kafka消费者
    /// </summary>
    public class KafkaConsumer : MQConsumerBase
    {
        private readonly KafkaConsumerConfig config;
        /// <summary>
        /// kafka消费者
        /// </summary>
        /// <param name="kcc"></param>
        /// <param name="handles"></param>
        public KafkaConsumer(KafkaConsumerConfig kcc, params IMessageHandler[] handles) : base(handles?.ToList())
        {
            this.config = kcc;
        }
        /// <inheritdoc/>
        protected override void ConnectToListen()
        {
            var config = new ConsumerConfig
            {
                GroupId = this.config.GroupId,
                ClientId = this.config.ClientId,
                BootstrapServers = this.config.Broker,
                AutoOffsetReset = AutoOffsetReset.Earliest
            };
            var consumer = new ConsumerBuilder<Ignore, string>(config).Build();
            consumer.Subscribe(this.config.Topic);

            StartConsumeAsync(consumer);
        }
        private void StartConsumeAsync(IConsumer<Ignore, string> consumer)
        {
            Task.Run(() =>
            {
                while (true)
                {
                    var msg = consumer.Consume();
                    HandleMessage(msg.Message.Value);
                }
            });
        }
        private Task HandleMessage(string message)
        {
            return Task.Run(() =>
            {
                JObject jsonObject = JObject.Parse(message);
                var typeString = jsonObject.GetString("_TypeString");
                //异步处理消息，（不能将参数[e]传入方法，handle方法是异步的，此方法return之后[e]随时可能被重新赋值，导致丢失消息数据）
                var handlers = GetMatchHandlers(typeString);
                if (handlers != null || handlers.Any())
                {
                    using (var ts = new TransactionScope())
                    {
                        foreach (var handler in handlers)
                        {
                            if (!handler.Handle(message))
                            {
                                throw new ApplicationException($"执行MessageHandler返回失败[{handler.GetType().FullName}]");
                            }
                        }
                        //res = true;
                        ts.Complete();
                    }
                }
            });
        }
    }
    /// <summary>
    /// kafka消费者配置
    /// </summary>
    public class KafkaConsumerConfig
    {
        /// <summary>
        /// 客户端ID
        /// </summary>
        public string ClientId { get; set; }
        /// <summary>
        /// 集群ID
        /// </summary>
        public string GroupId { get; set; }
        /// <summary>
        /// 主题
        /// </summary>
        public string Topic { get; set; }
        /// <summary>
        /// 工作节点
        /// </summary>
        public string Broker { get; set; }
    }
}
